//package org.lk.conver;
//
//import cn.hutool.db.Db;
//import cn.hutool.db.DbUtil;
//import cn.hutool.db.Entity;
//import cn.hutool.db.dialect.impl.MysqlDialect;
//import cn.hutool.db.ds.simple.SimpleDataSource;
//import cn.hutool.json.JSONUtil;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
//import org.lk.config.SpringUtil;
//
//import java.sql.Connection;
//import java.util.HashMap;
//import java.util.Map;
//
///**
// * @author: lk
// * @since: mysql接收函数
// */
//public class MysqlSinkFunction extends RichSinkFunction<String> {
//    Connection connection;
//    Db use;
////    FlinkEntity flinkEntity;
//
//    public MysqlSinkFunction() {
//
//    }
//
//    public MysqlSinkFunction(String flinkEntity) {
//        this.flinkEntity = JSONUtil.toBean(flinkEntity, FlinkEntity.class);
//
//    }
//
//    /**
//     * 这里可以做打开连接操作
//     */
//    @Override
//    public void open(Configuration parameters) throws Exception {
//        super.open(parameters);
//        SimpleDataSource simpleDataSource = new SimpleDataSource("jdbc:mysql://172.16.8.50:3306/flink_cdc_test2", "root",
//                "oceanDatum", "com.mysql.cj.jdbc.Driver");
//        connection = simpleDataSource.getConnection();
//        use = DbUtil.use(simpleDataSource, new MysqlDialect());
//    }
//
//    /**
//     * 这里可以进行真正的处理
//     */
//
//    @Override
//    public void invoke(String value, Context context) throws Exception {
//        HashMap hashMap = JSONUtil.toBean(value, HashMap.class);
//        String tsSec = String.valueOf(hashMap.get("tsSec"));
//        Integer pos = (Integer) hashMap.get("pos");
//        String file = (String) hashMap.get("file");
//        flinkEntity.setStartupTimestampMillis(tsSec);
//        flinkEntity.setSavePoint(pos);
//        flinkEntity.setBinlogFile(file);
//        Map<String, Object> data = (Map<String, Object>) hashMap.get("data");
//        String type = (String) hashMap.get("type");
//        String table = (String) hashMap.get("table");
//        if (type.equals("insert")) {
//            Entity record = new Entity();
//            record.setTableName(table);
//            data.forEach((key, values) -> {
//                record.put(key, values);
//            });
//            try {
//                use.insert(record);
//            } catch (Exception e) {
//                use.del(record);
//                use.insert(record);
//            }
//
//        } else if (type.equals("update")) {
//            Map<String, Object> updateBeforeData = (Map<String, Object>) hashMap.get("updateBeforeData");
//            Entity record = new Entity();
//            record.setTableName(table);
//            data.forEach((key, values) -> record.put(key, values));
//            Entity where = new Entity();
//            updateBeforeData.forEach((key, values) -> where.put(key, values));
//            use.update(record, where);
//        } else if (type.equals("delete")) {
//            Entity record = new Entity();
//            record.setTableName(table);
//            data.forEach((key, values) -> {
//                record.put(key, values);
//            });
//
//            use.del(record);
//        } else if (type.equals("select")) {
//            Entity record = new Entity();
//            record.setTableName(table);
//            data.forEach((key, values) -> record.put(key, values));
//            use.insert(record);
//
//        }
//        System.out.println(value);
//
//    }
//
//
//    /**
//     * 这里可以做关闭连接操作
//     *
//     * @throws Exception
//     */
//    @Override
//    public void close() throws Exception {
//        super.close();
//        use.closeConnection(connection);
//        FlinkEntityMapper bean = SpringUtil.getBean(FlinkEntityMapper.class);
//        flinkEntity.setHasFirstSync(true);
//        bean.updateById(flinkEntity);
//    }
//
//
//}
